package cn.ecpark.service.kafka.api;

import cn.ecpark.service.kafka.util.Topic;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;


@Slf4j
@RestController
@RequestMapping("/kafka")
public class KafkaProducerController {

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    // 发送消息
    @GetMapping("/normal/{message}")
    public String sendMessage1(@PathVariable("message") String normalMessage) {
        kafkaTemplate.send(Topic.NAME, normalMessage);
        return normalMessage;
    }


    //@Transactional  //如果kafka开启了事务，要加@Transactional注解
    @GetMapping("/callback/{message}")
    public String sendMessage1AndCallback(@PathVariable("message") String normalMessage) {
        kafkaTemplate.send(Topic.NAME, normalMessage).addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable ex) {
                log.info("发送失败：", ex.getMessage());
            }
            @Override
            public void onSuccess(SendResult<String, Object> result) {
                RecordMetadata recordMetadata = result.getRecordMetadata();
                log.info("发送成功,topic:{},partition:{},offset:{}",
                        recordMetadata.topic(),recordMetadata.partition(),recordMetadata.offset());
            }
        });

        return normalMessage;
    }


    //事务消息
    //@Transactional  //如果kafka开启了事务，要加@Transactional注解
    @GetMapping("/transaction/{message}")
    public void sendMessageTransaction(@PathVariable("message") String normalMessage) {

        //声明事务，后面报错消息发不出去;
        kafkaTemplate.executeInTransaction( operations -> {
            operations.send(Topic.NAME, normalMessage);
            throw new RuntimeException("fail");
        });

        //不声明事务，后面报错，但前面消息已经发出去了;
//        kafkaTemplate.send(Topic.NAME, normalMessage).addCallback(success -> {
//            log.info("发送成功,topic:{},partition:{},offset:{}",
//                    success.getRecordMetadata().topic(),success.getRecordMetadata().partition(),success.getRecordMetadata().offset()
//                    );
//        }, failue -> {
//            log.info("发送失败：", failue.getMessage());
//        });
//        throw new RuntimeException("fail");

    }

}
